package com.yutian.socketnio.client;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

/**
 * @author zhoulei7
 * @create 2020-09-14
 **/
public class Client {
    private static Logger log = LoggerFactory.getLogger(Client.class);

    public static void connectToServer(String ServerIp, int ServerPort, String greet) {
        SocketChannel sc = null;
        Selector selector = null;
        try {
            //创建客户端socket建立连接，指定服务器地址和端口
            // 打开socket通道
            sc = SocketChannel.open();
            //设置为非阻塞
            sc.configureBlocking(false);
            //连接服务器地址和端口
            sc.connect(new InetSocketAddress(ServerIp, ServerPort));
            //打开选择器
            selector = Selector.open();
            //注册连接服务器socket的动作
            sc.register(selector, SelectionKey.OP_CONNECT);

            while (!Thread.currentThread().isInterrupted()) {
                //选择一组键，其相应的通道已为 I/O 操作准备就绪。
                try {
                    //等待需要处理的新事件；阻塞将一直持续到下一个传入事件
                    selector.select();
                } catch (IOException ex) {
                    log.error(ex.getMessage(), ex);
                    //handle exception
                    break;
                }
                //返回此选择器的已选择键集。
                Set<SelectionKey> keys = selector.selectedKeys();
                log.info("keys=" + keys.size());
                Iterator<SelectionKey> keyIterator = keys.iterator();
                while (keyIterator.hasNext()) {
                    SelectionKey key = keyIterator.next();
                    if (!key.isValid() /*&& key.isConnectable()*/) {
                        continue;
                    }
                    handleSelectionKey(sc, selector, key);
                    //该事件已经处理，可以丢弃
                    keyIterator.remove();
                }
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }

    }

    private static void handleSelectionKey(SocketChannel sc, Selector selector, SelectionKey key) {
        // 判断此通道上是否正在进行连接操作。 顺序：连接，注册写事件 - 写，注册读事件 - 读，注册写事件
        if (key.isConnectable()) {
            connect(sc, selector);
        } else if (key.isWritable()) { //写数据
            write(sc, selector);
        } else if (key.isReadable()) {//读取数据
            read(sc, selector, key);
        }
    }

    private static void read(SocketChannel sc, Selector selector, SelectionKey key) {
        log.info("receive message...");
        SocketChannel client = (SocketChannel) key.channel();
        ByteBuffer readBuffer = ByteBuffer.allocate(1024);//调整缓存的大小可以看到打印输出的变化
        //将缓冲区清空以备下次读取
        //readBuffer.clear();
        int num = 0;
        try {
            num = client.read(readBuffer);
        } catch (IOException e) {
            // 对方断开连接，FD_READ仍然起作用，但读取异常，所以从Selector中取消注册  TODO 未确认
            // The remote forcibly closed the connection, cancel
            // the selection key and close the channel.
            closeChannel(key, client);
            log.error("client.read.error", e);
            return;
        }
        if(-1 == num){
            // 客户端主动关闭连接，服务器端并不会出现远程主机强迫关闭一个现有的连接这样的异常，而是会一直出现可读事件，一直返回-1。已验证！
            closeChannel(key, client);
            log.error("client.read.-1");
            return;
        }
        try {
            // 模拟业务逻辑耗时
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String str = new String(readBuffer.array(), 0, num);
        log.info("read:" + str);
        //注册读操作，下一次读取
        try {
            sc.register(selector, SelectionKey.OP_WRITE);
        } catch (ClosedChannelException e) {
            log.error(e.getMessage(), e);
        }
    }

    private static void closeChannel(SelectionKey key, SocketChannel client) {
        key.cancel();
        try {
            client.close();
        } catch (IOException ex) {
            log.error(ex.getMessage(), ex);
        }
    }

    private static void write(SocketChannel sc, Selector selector) {
        try {
            // 模拟业务逻辑耗时
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String message = "Hi，我来自客户端端的问候！";
        log.info("write:" + message);
        ByteBuffer sendBuffer = ByteBuffer.allocate(1024);//调整缓存的大小可以看到打印输出的变化
        sendBuffer.clear();
        sendBuffer.put(message.getBytes());
        //将缓冲区各标志复位,因为向里面put了数据标志被改变要想从中读取数据发向服务器,就要复位
        sendBuffer.flip();
        try {
            sc.write(sendBuffer);

            //注册写操作,每个chanel只能注册一个操作，最后注册的一个生效
            //如果你对不止一种事件感兴趣，那么可以用“位或”操作符将常量连接起来
            //int interestSet = SelectionKey.OP_READ | SelectionKey.OP_WRITE;
            sc.register(selector, SelectionKey.OP_READ);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    private static void connect(SocketChannel sc, Selector selector) {
        try {
            if (sc.finishConnect()) {
                sc.register(selector, SelectionKey.OP_WRITE);
                log.info("server connected...");
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }
}
